package com.tiger.cloud.common.redis.queue;

import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.tiger.cloud.util.ThreadUtils;

public class TaskMonitor {

	private RedisQueue redisQueue;
	private static final int DEFAULT_INTERVAL = 3 * 60;
	private int interval = DEFAULT_INTERVAL;
	private AtomicBoolean started = new AtomicBoolean(false);
	private AtomicBoolean closed = new AtomicBoolean(false);
	private ScheduledExecutorService scheduler = null;

	public TaskMonitor(RedisQueue redisQueue) {
		this.redisQueue = redisQueue;
	}

	public void setRedisQueue(RedisQueue redisQueue) {
		this.redisQueue = redisQueue;
	}

	public void setInterval(int interval) {
		this.interval = interval;
	}

	public void start() {
		if (!started.compareAndSet(false, true)) {
			throw new IllegalStateException("this TaskMonitor already start!");
		}
		if (scheduler == null) {
			scheduler = Executors.newSingleThreadScheduledExecutor();
		}
		scheduler.scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				try {
					redisQueue.checkExpireMessage();
				} catch (Exception e) {
					// handle Exception
				}
			}
		}, 5, interval, TimeUnit.SECONDS);
	}

	public synchronized void close() throws IOException {
		if (!closed.compareAndSet(false, true)) {
			throw new IllegalStateException("this TaskMonitor already close!");
		}
		if (scheduler != null) {
			ThreadUtils.shutdownAndAwaitTermination(scheduler, 5, TimeUnit.SECONDS);
		}
	}
}
